package com.awesome.spring.function.process;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @program: awesome-spring
 * @description: 对数据进行更复杂的处理，比如访问时间上下文、定时器、侧输出流等，属于 流处理（Processing）操作。
 * @author: Alex Wu
 * @createDate: 2025-03-20 15:42
 **/
public class BehaviorLogProcessFunc extends ProcessFunction<String, String> {

    // 定义一个状态，用于存储时间戳
//    private transient ValueState<Long> timerState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化状态
        super.open(parameters);
        /*ValueStateDescriptor<Long> descriptor = new ValueStateDescriptor<>("timerState", Long.class);
        timerState = getRuntimeContext().getState(descriptor);*/
    }

    @Override
    public void processElement(String value, Context context, Collector<String> collector) throws Exception {
        /*if (record == null) {
            return;
        }
        String newsId = record.getId();

        long start = System.currentTimeMillis();
        // 获取标题生成结果
        String titleContent = this.getNewsAiResult(record.getTitle(), AI_MODEL_NEWS_TITLE);
        record.setTitle(titleContent);
        long end1 = System.currentTimeMillis();
        long titleTransformCost = end1 - start;
        log.info("HTTP getNewsTitleContent Finished: id {}, cost {} ms", newsId, titleTransformCost);

        // 获取摘要生成结果
        String summaryContent = this.getNewsAiResult(record.getContent(), AI_MODEL_NEWS_SUMMARY);
        record.setSummaryContent(summaryContent);
        long contentTransformCost = System.currentTimeMillis() - end1;
        log.info("HTTP getNewsSummaryContent Finished: id {}, cost {} ms", newsId, contentTransformCost);

        if (StringUtils.isBlank(record.getTitle())
                || StringUtils.isBlank(record.getSummaryContent())) {
            record.setTitleTransformCost(titleTransformCost);
            record.setContentTransformCost(contentTransformCost);
        }

        collector.collect(record);*/


        // 获取当前事件时间
        long currentTimestamp = context.timestamp() != null ? context.timestamp() : System.currentTimeMillis();

        // 计算 10 秒后的时间戳
        long timerTimestamp = currentTimestamp + 10_000;
//        timerState.update(timerTimestamp);

        // 注册事件时间定时器
//        context.timerService().registerEventTimeTimer(timerTimestamp);

        // 立即输出一条日志
        collector.collect("Received: " + value + " at " + currentTimestamp);
    }

}
